package io.gitee.kenewstar.dbmq.handler;

import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUtil;
import io.gitee.kenewstar.dbmq.ConsumerRegisterInfo;
import io.gitee.kenewstar.dbmq.ConsumerTask;
import io.gitee.kenewstar.dbmq.DbMqConfigProperties;
import io.gitee.kenewstar.dbmq.thread.WorkThreadPool;
import io.gitee.kenewstar.dbmq.data.DbDataHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;

/**
 * @author kenewstar
 */
public class PullMessageHandler extends DbMqCoreHandler implements Runnable {

    private static final Logger LOGGER = LoggerFactory.getLogger(PullMessageHandler.class);

    private List<ConsumerTask> taskList;
    private List<ConsumerRegisterInfo> registerList;
    private WorkThreadPool workThreadPool;

    public PullMessageHandler() {}

    public PullMessageHandler(DbDataHandler dbDataHandler,
                              DbMqConfigProperties dbMqConfigProperties,
                              WorkThreadPool workThreadPool) {
        this.dbDataHandler = dbDataHandler;
        this.dbMqConfigProperties = dbMqConfigProperties;
        this.workThreadPool = workThreadPool;
    }

    public void setTaskList(List<ConsumerTask> taskList) {
        this.taskList = taskList;
    }

    @Override
    public void run() {
        this.registerList = pullRegisterUpdate(taskList);

        this.checkSendHeartbeat();

        this.workThreadPool.pullMessageTaskRunning();
    }

    private void checkSendHeartbeat() {

        final DateTime current = DateUtil.date();
        final List<Long> ids = registerList.stream().filter(x -> matchConsumer(taskList, x)).filter(x -> {
            int heartbeatInterval = dbMqConfigProperties.getHeartbeatInterval();
            if (dbMqConfigProperties.getHeartbeatInterval() < dbMqConfigProperties.getPeriod()) {
                heartbeatInterval = dbMqConfigProperties.getPeriod();
            }
            final DateTime nextHeartbeat = DateUtil.offsetMillisecond(x.getHeartbeatTime(), heartbeatInterval);
            return DateUtil.compare(current, nextHeartbeat) >= 0;
        }).map(ConsumerRegisterInfo::getId).collect(Collectors.toList());

        this.dbDataHandler.sendHeartbeat(ids);

    }

    private boolean matchConsumer(List<ConsumerTask> taskList, ConsumerRegisterInfo info) {
        return taskList.stream().anyMatch(y -> Objects.equals(y.getId(), info.getId()));
    }



}
